package com.changan.examples;

import com.changan.utils.FlinkRestApiUtil;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
public class CancelWithSavepoint {
    private static final String FLINK_HOST = "";
    private static final String FLINK_PORT = "";

    public static void main(String[] args) throws JsonProcessingException {
        ObjectMapper objectMapper = new ObjectMapper();

        String flinkJob = FlinkRestApiUtil.getListOfFlinkJob(FLINK_HOST, FLINK_PORT);
        System.out.println(flinkJob);

        JsonNode jsonNode = objectMapper.readTree(flinkJob);
        String jobId = jsonNode.get("jobs").get(0).get("id").asText();
        System.out.println(jobId);
        CancelArgs stopArgs = new CancelArgs();
        stopArgs.setCancelJob(false);
        stopArgs.setTargetDirectory("hdfs://nameservice1/tmp/yarnjobs/");
        String response = FlinkRestApiUtil.triggerSavepoint(
                FLINK_HOST,
                FLINK_PORT,
                jobId,
                stopArgs
        );
        //{"request-id":"d235b264d99618ca3714c812c47525a8"}
        System.out.println(response);
        JsonNode requestIdNode = objectMapper.readTree(response.toString());
        String requestId = requestIdNode.get("request-id").asText();
        String statusOfSaOp = FlinkRestApiUtil.getStatusOfSaOp(FLINK_HOST, FLINK_PORT, jobId, requestId);
        System.out.println(statusOfSaOp);

    }

    @Getter
    @Setter
    @NoArgsConstructor
    @AllArgsConstructor
    public static class CancelArgs {
        @JsonProperty("target-directory")
        private String targetDirectory;

        @JsonProperty("cancel-job")
        private Boolean cancelJob;
    }

}
